package org.javaboy.redis_stream_demo.service;

import jakarta.annotation.PostConstruct;
import org.javaboy.redis_stream_demo.config.RedisQueueConfig;
import org.javaboy.redis_stream_demo.model.User;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.hash.ObjectHashMapper;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import java.time.Duration;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 手动 ACK 消费配置
 */
//@Configuration
public class AutoAckConsumerConfig {
    @Autowired
    RedisTemplate redisTemplate;

    /**
     * 如果是分组消费，则必须提前创建好消费组，否则会报错
     * 可以通过命令去创建
     */
    @PostConstruct
    public void init() {
        StreamOperations streamOperations = redisTemplate.opsForStream();
        try {
            //创建两个消费组
            streamOperations.createGroup(RedisQueueConfig.QUEUE_NAME, "group-a");
            streamOperations.createGroup(RedisQueueConfig.QUEUE_NAME, "group-b");
        } catch (Exception e) {
//            throw new RuntimeException(e);
        }
    }

    @Bean
    StreamMessageListenerContainer<String, ObjectRecord<String, User>> stringObjectRecordStreamMessageListenerContainer() {
        AtomicInteger index = new AtomicInteger(1);
        int processors = Runtime.getRuntime().availableProcessors();
        ThreadPoolExecutor executor = new ThreadPoolExecutor(processors, processors * 2, 0, TimeUnit.SECONDS, new LinkedBlockingDeque<>(), runnable -> {
            Thread thread = new Thread(runnable);
            thread.setName("consumer:" + index.getAndIncrement());
            thread.setDaemon(true);
            return thread;
        });
        //消费容器的属性配置
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, User>> containerOptions = StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
                //一次拉取多少条数据
                .batchSize(10)
                .executor(executor)
                .keySerializer(redisTemplate.getKeySerializer())
                .hashValueSerializer(redisTemplate.getHashValueSerializer())
                .hashKeySerializer(redisTemplate.getHashKeySerializer())
                //拉取消息的等待超时时间
                .pollTimeout(Duration.ofSeconds(10))
                .errorHandler(new ConsumerExceptionHandler())
                .objectMapper(new ObjectHashMapper())
                .targetType(User.class)
                .build();
        //创建消费容器
        StreamMessageListenerContainer<String, ObjectRecord<String, User>> container = StreamMessageListenerContainer.create(redisTemplate.getConnectionFactory(), containerOptions);
        container.receiveAutoAck(Consumer.from("group-a", "consumer-1"), StreamOffset.create(RedisQueueConfig.QUEUE_NAME, ReadOffset.lastConsumed()), new AsyncStreamListener("group-a", "consumer-1",redisTemplate));
        container.receiveAutoAck(Consumer.from("group-a", "consumer-2"), StreamOffset.create(RedisQueueConfig.QUEUE_NAME, ReadOffset.lastConsumed()), new AsyncStreamListener("group-a", "consumer-2",redisTemplate));
        container.receiveAutoAck(Consumer.from("group-b", "consumer-3"), StreamOffset.create(RedisQueueConfig.QUEUE_NAME, ReadOffset.lastConsumed()), new AsyncStreamListener("group-b", "consumer-3",redisTemplate));
        //启动消费容器
        container.start();
        return container;
    }
}
